package com.intct.flink.cdc;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author gufg
 * @since 2025-07-01 15:52
 */
public class FlinkCDC1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // MySqlSource定义
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("jd-node")
                .port(3306)
                .username("root")
                .password("999999")
                .databaseList("travel")
                .tableList("travel.book")
                // 反序列化
                .deserializer(new JsonDebeziumDeserializationSchema())
                // 指定读数据的位置
                /*
                    配置选项scan.startup.mode指定 MySQL CDC 使用者的启动模式。有效枚举包括：
                    initial （默认）：在第一次启动时对受监视的数据库表执行初始快照，并继续读取最新的 binlog。
                    earliest-offset：跳过快照阶段，从可读取的最早 binlog 位点开始读取
                    latest-offset：首次启动时，从不对受监视的数据库表执行快照， 连接器仅从 binlog 的结尾处开始读取，这意味着连接器只能读取在连接器启动之后的数据更改。
                    specific-offset：跳过快照阶段，从指定的 binlog 位点开始读取。位点可通过 binlog 文件名和位置指定，或者在 GTID 在集群上启用时通过 GTID 集合指定。
                    timestamp：跳过快照阶段，从指定的时间戳开始读取 binlog 事件。
                 */
                .startupOptions(StartupOptions.initial())
                .build();

        // 获取数据源数据
        DataStreamSource<String> sourceDS = env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-source");

        SingleOutputStreamOperator<String> mapDS = sourceDS.map(new MapFunction<String, String>() {
            @Override
            public String map(String jsonStr) throws Exception {
                JSONObject jsonObj = JSONObject.parseObject(jsonStr);
                jsonObj.remove("source");
                jsonObj.remove("ts_ms");
                jsonObj.remove("transaction");

                return JSON.toJSONString(jsonObj);
            }
        });


        // Kafka sink
        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                .setBootstrapServers("cdh-node:9092")
                .setRecordSerializer(KafkaRecordSerializationSchema.<String>builder()
                        .setTopic("intct81")
                        .setValueSerializationSchema(new SimpleStringSchema())
                        .build())
                .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
                .build();

        // 打印
        mapDS.sinkTo(kafkaSink);

        env.execute();
    }
}
